package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV1;
import com.group1.realtime.bean.KeywordBean;
import com.group1.realtime.common.Constant;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import com.group1.realtime.util.IkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

public class Dws_TrafficKeywordPageViewWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_TrafficKeywordPageViewWindow().init(
                3099,
                2,
                "Dws_TrafficKeywordPageViewWindow",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<JSONObject> keywordStream =findKeyword(stream);
        SingleOutputStreamOperator<Tuple2<String, Long>> splitStream = splitKeyword(keywordStream);
        SingleOutputStreamOperator<KeywordBean> aggStream = windowAndAgg(splitStream);
        writeClikeHouse(aggStream);

    }

    private void writeClikeHouse(SingleOutputStreamOperator<KeywordBean> aggStream) {
        aggStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_keyword_page_view_window",KeywordBean.class));
    }

    private SingleOutputStreamOperator<KeywordBean> windowAndAgg(SingleOutputStreamOperator<Tuple2<String, Long>> splitStream) {
        return splitStream
                .map(new MapFunction<Tuple2<String, Long>, KeywordBean>() {
                         @Override
                         public KeywordBean map(Tuple2<String, Long> s) throws Exception {
                             return  KeywordBean.builder()
                                     .word(s.f0)
                                     .source("search")
                                     .wordCt(1L)
                                     .ts(s.f1)
                                     .build();
                         }
                     }
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<KeywordBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean,ts)->bean.getTs())
                )
                .keyBy(bean->bean.getWord())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<KeywordBean>() {
                            @Override
                            public KeywordBean reduce(KeywordBean t1, KeywordBean t2) throws Exception {
                                t1.setWordCt(t1.getWordCt() + t2.getWordCt());
                                return t1;
                            }
                        },
                        new ProcessWindowFunction<KeywordBean, KeywordBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context context,
                                                Iterable<KeywordBean> iterable,
                                                Collector<KeywordBean> out) throws Exception {
                                KeywordBean next = iterable.iterator().next();

                                next.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                next.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));

                                next.setTs(context.currentProcessingTime());

                                out.collect(next);
                            }
                        });
    }


    private SingleOutputStreamOperator<Tuple2<String, Long>> splitKeyword(SingleOutputStreamOperator<JSONObject> keywordStream) {
//        keywordStream.print();
        return keywordStream
                .flatMap(new FlatMapFunction<JSONObject, Tuple2<String,Long>>() {
                    @Override
                    public void flatMap(JSONObject jsonObject, Collector<Tuple2<String,Long>> out) throws Exception {
                        JSONObject page = jsonObject.getJSONObject("page");
                        Long ts = jsonObject.getLong("ts");
                        String keywords = page.getString("item");
                        ArrayList<String> words = IkUtil.split(keywords);
                        for (String word : words) {
                            out.collect(Tuple2.of(word,ts));
                        }
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> findKeyword(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        JSONObject page = jsonObject.getJSONObject("page");
                        String last_page_id = page.getString("last_page_id");
                        String item_type = page.getString("item_type");
                        String item = page.getString("item");

                        return ("home".equals(last_page_id) || "search".equals(last_page_id))
                                && "keyword".equals(item_type)
                                && item!=null;
                    }
                });
    }
}
